package de.distMLP.messages;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;

import de.mlp_distributed.mlp.math.mahout.Vector;

public class VectorMessage extends HamaMessage {

	private Vector vector;

	private final Kryo kryo = new Kryo();

	byte[] serializedArray = null;

	public VectorMessage() {
		super();
	}

	public VectorMessage(final Vector v) throws IOException {
		this.vector = v;
		this.serializedArray = this.serialize();
	}

	public byte[] serialize() throws IOException {
		final ByteArrayOutputStream out = new ByteArrayOutputStream();

		final Output fileOutput = new Output(out);
		this.kryo.writeObjectOrNull(fileOutput, this.vector, Vector[].class);

		fileOutput.flush();
		fileOutput.close();
		out.flush();
		out.close();
		return out.toByteArray();
	}

	public void deserialize(final byte[] input) throws IOException {
		final ByteArrayInputStream in = new ByteArrayInputStream(input);
		final Input fileOutput = new Input(in);
		this.vector = this.kryo.readObjectOrNull(fileOutput, Vector.class);
	}

	@Override
	public void readFields(final DataInput in) throws IOException {
		final int size = in.readInt();
		final byte[] data = new byte[size];
		in.readFully(data);
		this.deserialize(data);
	}

	@Override
	public void write(final DataOutput out) throws IOException {
		if (this.serializedArray == null) {
			this.serializedArray = this.serialize();
		}
		out.writeInt(this.serializedArray.length);
		out.write(this.serializedArray);
	}

	public Vector getVector() {
		return this.vector;
	}

	public void setVector(final Vector vector) {
		this.vector = vector;
	}

}
